-
Notifications
You must be signed in to change notification settings - Fork 4
기타 설치된 패키지
ChangSoo Choi edited this page Jul 6, 2023
·
4 revisions
✔️ 기타 패키지
- django-apscheduler (https://pypi.org/project/django-apscheduler/)
- 백그라운드에서 자동으로 작업 실행
- 휴면계정 삭제와 쿠팡링크 요청에 사용됨
- 어드민 페이지에서 로그 확인 및 관리 가능
- django-taggit(+taggit-serializer) (https://pypi.org/project/django-taggit/)
- 태그 기능을 drf에 적용함
- 게시글 작성시 태그 추가가 가능함
- 사용자가 게시글 검색시 태그로 검색기능에 연동됨
- ratelimiter (https://pypi.org/project/ratelimiter/)
- 쿠팡 API 지침에 따라 1분에 최대 50번만 호출되므로 해당 패키지를 통해 서버 전체적으로 API 요청을 1분에 50번만 되도록 제한함
- Roboflow
- 머신러닝을 진행하는데 사용되며 사물인식 기능을 수행함
- Scipy
- SVD와 벡터 유사도 연산을 위해 사용한 패키지이며, 이를 통해 추천 알고리즘을 구현함
- Pandas
- 장고ORM데이터를 Scipy등의 라이브러리가 사용할 수 있게 dataframe으로 만들고 dataframe을 조작하기 위한 패키지
💻 code
django-apscheduler | coupang.py
class CoupangManage:
DOMAIN = "https://api-gateway.coupang.com"
# HMAC서명 생성
def generateHmac(self, method, url, secretKey, accessKey):
path, *query = url.split("?")
current_datetime = datetime.utcnow().strftime("%y%m%dT%H%M%SZ")
message = current_datetime + method + path + (query[0] if query else "")
signature = hmac.new(
bytes(secretKey, "utf-8"), message.encode("utf-8"), hashlib.sha256
).hexdigest()
return "CEA algorithm=HmacSHA256, access-key={}, signed-date={}, signature={}".format(
accessKey, current_datetime, signature
)
def get_productsdata(self, request_method, authorization, url):
response = requests.request(
method=request_method,
url=url,
headers={
"Authorization": authorization,
"Content-Type": "application/json;charset=UTF-8",
},
)
retdata = json.dumps(response.json(), indent=4).encode("utf-8")
jsondata = json.loads(retdata)
data = jsondata["data"]
productdata = data["productData"]
return productdata
def get_products_by_keyword(self, keyword, limit=10, subld="cookaicookai"):
request_method = "GET"
method = "GET"
query = {"keyword": keyword, "limit": limit, "subId": subld}
encoded_query = urlencode(query)
# API 요청 URL
request_url = (
"/v2/providers/affiliate_open_api/apis/openapi/products/search?{}".format(
encoded_query
)
)
# HMAC 생성
authorization = self.generateHmac(method, request_url, SECRET_KEY, ACCESS_KEY)
# 요청 URL
full_url = "{}{}".format(self.DOMAIN, request_url)
# 상품 데이터 받기
product_data = self.get_productsdata(request_method, authorization, full_url)
# 딕셔너리 생성
products = [
{
"product_url": product["productUrl"],
"product_image_url": product["productImage"],
"product_price": product["productPrice"],
}
for product in product_data
]
return products
django-taggit | coupang.py
class CoupangManage:
DOMAIN = "https://api-gateway.coupang.com"
# HMAC서명 생성
def generateHmac(self, method, url, secretKey, accessKey):
path, *query = url.split("?")
current_datetime = datetime.utcnow().strftime("%y%m%dT%H%M%SZ")
message = current_datetime + method + path + (query[0] if query else "")
signature = hmac.new(
bytes(secretKey, "utf-8"), message.encode("utf-8"), hashlib.sha256
).hexdigest()
return "CEA algorithm=HmacSHA256, access-key={}, signed-date={}, signature={}".format(
accessKey, current_datetime, signature
)
def get_productsdata(self, request_method, authorization, url):
response = requests.request(
method=request_method,
url=url,
headers={
"Authorization": authorization,
"Content-Type": "application/json;charset=UTF-8",
},
)
retdata = json.dumps(response.json(), indent=4).encode("utf-8")
jsondata = json.loads(retdata)
data = jsondata["data"]
productdata = data["productData"]
return productdata
def get_products_by_keyword(self, keyword, limit=10, subld="cookaicookai"):
request_method = "GET"
method = "GET"
query = {"keyword": keyword, "limit": limit, "subId": subld}
encoded_query = urlencode(query)
# API 요청 URL
request_url = (
"/v2/providers/affiliate_open_api/apis/openapi/products/search?{}".format(
encoded_query
)
)
# HMAC 생성
authorization = self.generateHmac(method, request_url, SECRET_KEY, ACCESS_KEY)
# 요청 URL
full_url = "{}{}".format(self.DOMAIN, request_url)
# 상품 데이터 받기
product_data = self.get_productsdata(request_method, authorization, full_url)
# 딕셔너리 생성
products = [
{
"product_url": product["productUrl"],
"product_image_url": product["productImage"],
"product_price": product["productPrice"],
}
for product in product_data
]
return products
ratelimiter|coupang.py
class CoupangManage:
DOMAIN = "https://api-gateway.coupang.com"
# HMAC서명 생성
def generateHmac(self, method, url, secretKey, accessKey):
path, *query = url.split("?")
current_datetime = datetime.utcnow().strftime("%y%m%dT%H%M%SZ")
message = current_datetime + method + path + (query[0] if query else "")
signature = hmac.new(
bytes(secretKey, "utf-8"), message.encode("utf-8"), hashlib.sha256
).hexdigest()
return "CEA algorithm=HmacSHA256, access-key={}, signed-date={}, signature={}".format(
accessKey, current_datetime, signature
)
def get_productsdata(self, request_method, authorization, url):
response = requests.request(
method=request_method,
url=url,
headers={
"Authorization": authorization,
"Content-Type": "application/json;charset=UTF-8",
},
)
retdata = json.dumps(response.json(), indent=4).encode("utf-8")
jsondata = json.loads(retdata)
data = jsondata["data"]
productdata = data["productData"]
return productdata
def get_products_by_keyword(self, keyword, limit=10, subld="cookaicookai"):
request_method = "GET"
method = "GET"
query = {"keyword": keyword, "limit": limit, "subId": subld}
encoded_query = urlencode(query)
# API 요청 URL
request_url = (
"/v2/providers/affiliate_open_api/apis/openapi/products/search?{}".format(
encoded_query
)
)
# HMAC 생성
authorization = self.generateHmac(method, request_url, SECRET_KEY, ACCESS_KEY)
# 요청 URL
full_url = "{}{}".format(self.DOMAIN, request_url)
# 상품 데이터 받기
product_data = self.get_productsdata(request_method, authorization, full_url)
# 딕셔너리 생성
products = [
{
"product_url": product["productUrl"],
"product_image_url": product["productImage"],
"product_price": product["productPrice"],
}
for product in product_data
]
return products
views.py
class LinkPlusView(APIView):
permission_classes = [permissions.IsAuthenticated]
def get(self, request, article_id):
# 로그인된 사용자의 Fridge 조회
user_fridge_ingredients = Fridge.objects.filter(user=request.user).values_list(
"ingredient", flat=True
)
# article_id와 연결된 RecipeIngredient 조회
recipe_ingredients = RecipeIngredient.objects.filter(
article_id=article_id
).values_list("ingredient", flat=True)
# 사용자의 Fridge에 없는 Ingredient 찾기
missing_ingredients = set(recipe_ingredients) - set(user_fridge_ingredients)
# 없는 Ingredient와 연결된 IngredientLink 조회
# column_name+__in : 리스트 안에 지정한 문자열들 중에 하나라도 포함된 데이터를 찾을 때 사용
ingredient_links = IngredientLink.objects.filter(
ingredient__in=missing_ingredients
)
# 현재 시간과 5일 전 시간 구하기
now = timezone.now()
five_days_ago = now - timedelta(days=5)
# 존재하지 않는 IngredientLink 인스턴스 생성 및 저장
for ingredient in missing_ingredients:
ingredient_links = IngredientLink.objects.filter(ingredient=ingredient)
if not ingredient_links.exists():
save_coupang_links_to_ingredient_links(ingredient)
else:
# link가 등록된 재료라도 `updated_at`이 5일 이상 지났다면 최신화하기
existing_link = ingredient_links.first()
if (
existing_link.ingredient.updated_at is None
or existing_link.ingredient.updated_at <= five_days_ago
):
save_coupang_links_to_ingredient_links(ingredient)
# 다시 IngredientLink를 조회
ingredient_links = IngredientLink.objects.filter(
ingredient__in=missing_ingredients
)
# JSON 형태로 반환
serialized_links = IngredientLinkSerializer(ingredient_links, many=True)
return Response(serialized_links.data, status=status.HTTP_200_OK)